package com.ruoyi.search.listener;

import cn.throwx.canal.gule.model.CanalBinLogEvent;
import cn.throwx.canal.gule.model.CanalBinLogResult;
import cn.throwx.canal.gule.support.processor.BaseCanalBinlogEventProcessor;
import cn.throwx.canal.gule.support.processor.ExceptionHandler;
import com.alibaba.fastjson.JSONObject;
import com.ruoyi.common.core.constant.SecurityConstants;
import com.ruoyi.common.core.domain.R;
import com.ruoyi.common.core.exception.base.BaseException;
import com.ruoyi.search.bo.EsOperLog;
import com.ruoyi.search.constant.EsIndexEnum;
import com.ruoyi.system.api.RemoteLogService;
import com.ruoyi.system.api.domain.SysOperLog;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.IOException;

@Component
@Slf4j
public class OperLogCanalListener extends BaseCanalBinlogEventProcessor<EsOperLog> {

    @Autowired
    private RemoteLogService remoteLogService;
    @Autowired
    private RestHighLevelClient restHighLevelClient;

    /**
     * 插入订单，此时插入es
     */
    @Override
    protected void processInsertInternal(CanalBinLogResult<EsOperLog> result) {
        Long orderId = result.getPrimaryKey();

        EsOperLog data = result.getAfterData();

        IndexRequest request = new IndexRequest(EsIndexEnum.OPERLOG.value());
        request.id(String.valueOf(orderId));
        request.source(JSONObject.toJSONString(data), XContentType.JSON);
        try {
            IndexResponse indexResponse = restHighLevelClient.index(request, RequestOptions.DEFAULT);
            log.info(indexResponse.toString());
        } catch (IOException e) {
            e.printStackTrace();
            log.error(e.toString());
            throw new BaseException("保存操作日志信息异常", e.toString());
        }
    }

    /**
     * 更新订单，删除订单索引，再重新构建一个
     */
    @Override
    protected void processUpdateInternal(CanalBinLogResult<EsOperLog> result) {
        Long orderId = result.getPrimaryKey();
        EsOperLog data = result.getAfterData();
        UpdateRequest request = new UpdateRequest(EsIndexEnum.OPERLOG.value(), String.valueOf(orderId));
        request.doc(JSONObject.toJSONString(data), XContentType.JSON);
        request.docAsUpsert(true);
        try {
            UpdateResponse updateResponse = restHighLevelClient.update(request, RequestOptions.DEFAULT);
            log.info(updateResponse.toString());
        } catch (IOException e) {
            log.error(e.toString());
            throw new BaseException("更新操作日志信息异常",e.toString());
        }
    }

    @Override
    protected ExceptionHandler exceptionHandler() {
        return (CanalBinLogEvent event, Throwable throwable) -> {
            throw new BaseException("创建索引异常",throwable.toString());
        };
    }
}
